package com.app.function;

import com.alibaba.fastjson.JSONObject;
import com.common.GmallConfig;
import com.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2022/1/4 10:41
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection connection;


    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        connection.setAutoCommit(true);
    }

    //TODO:value:{"sinkTable":"dim_base_trademark","database":"gmall-flink","before":{"tm_name":"atguigu","id":12},"after":{"tm_name":"Atguigu","id":12},"type":"update","tableName":"base_trademark"}
    //TODO:SQL：upsert into db.tn(id,tm_name) values('...','...')
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {

        PreparedStatement preparedStatement = null;
        try {
            //获取SQL语句
            String sinkTable = value.getString("sinkTable");
            JSONObject after = value.getJSONObject("after");
            String upsertSql = getUpsertSql(sinkTable, after);

            //预编译SQL
            preparedStatement = connection.prepareStatement(upsertSql);

            //判断如果当前数据为更新操作，则先删除Redis中的数据
            if ("update".equals(value.getString("type"))){
                DimUtil.delRedisDimInfo(sinkTable.toUpperCase(),after.getString("id"));
            }

            //执行插入操作
            preparedStatement.executeUpdate();

            //提交
           /* connection.commit();*/
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            if (preparedStatement!=null){
                preparedStatement.close();
            }
        }

    }


    //TODO:data:{"tm_name":"Atguigu","id":12}
    //TODO:SQL：upsert into db.tn(id,tm_name,aa,bb) values('...','...','...','...')
    private String getUpsertSql(String sinkTable, JSONObject data) {
        Set<String> keySet = data.keySet();
        Collection<Object> values = data.values();

        StringBuffer buffer = new StringBuffer();
        buffer.append("upsert into ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append("(")
                .append(StringUtils.join(keySet, ","))
                .append(")values('")
                .append(StringUtils.join(values, "','"))
                .append("')");

        return buffer.toString();
    }
}
